package cn.ac.iie.ulss.indexer.worker;

import cn.ac.iie.ulss.indexer.metastore.MetastoreWrapper;
import cn.ac.iie.ulss.indexer.metric.DataIndexerMetricReportor;
import cn.ac.iie.ulss.indexer.mqproducer.RocketMQProducer;
import cn.ac.iie.ulss.indexer.runenvs.GlobalParas;
import cn.ac.iie.ulss.indexer.writer.DataSourceConfig;
import cn.ac.iie.ulss.indexer.writer.LuceneFileWriter;
import cn.ac.iie.ulss.indexer.writer.LuceneWriterUtil;
import com.alibaba.rocketmq.common.message.MessageExt;
import devmap.DevMap;
import java.io.ByteArrayInputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericArray;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.util.Utf8;
import org.apache.hadoop.hive.metastore.api.SFile;
import org.apache.hadoop.hive.metastore.api.SFileLocation;
import org.apache.hadoop.hive.metastore.api.SplitValue;
import org.apache.log4j.Logger;
import org.apache.lucene.index.IndexWriter;

public class DataDispatcher implements Runnable {

    public static Logger log = Logger.getLogger(DataDispatcher.class.getName());
    private ArrayBlockingQueue<MessageExt> buffer = null;
    private Schema docsSchema = null;
    private DatumReader<GenericRecord> docsReader = null;

    DataDispatcher(ArrayBlockingQueue<MessageExt> buffer) {
        this.buffer = buffer;
        this.docsSchema = GlobalParas.docsProtocol.getType(GlobalParas.docs);
        this.docsReader = new GenericDatumReader(this.docsSchema);
    }

    public static String getOriTopic(MessageExt msg) {
        return msg.getTopic().replaceFirst(new StringBuilder().append(GlobalParas.hostName).append("_").toString(), "");
    }

    @Override
    public void run() {
        MessageExt msg = null;
        while (true) {
            try {
                msg = (MessageExt) this.buffer.poll(8L, TimeUnit.MILLISECONDS);
            } catch (Exception ex) {
                log.error(ex, ex);
            }
            if (msg == null) {
                continue;
            }
            int docSetSize = 0;
            long bg = System.currentTimeMillis();
            ByteArrayInputStream docsbis = null;
            BinaryDecoder docsbd = null;
            GenericRecord docsRecord = null;
            IndexControler idxCtrl = null;
            GenericArray docSet = null;
            long file_id = -1L;
            docsbis = new ByteArrayInputStream(msg.getBody());
            log.info(msg.getBody());
            docsbd = new DecoderFactory().binaryDecoder(docsbis, null);
            docsRecord = new GenericData.Record(this.docsSchema);
            try {
                this.docsReader.read(docsRecord, docsbd);
                docSet = (GenericArray) docsRecord.get(GlobalParas.docs_set);    
                docSetSize = docSet.size();
                ((AtomicLong) GlobalParas.getdataStatics.get(msg.getTopic())).addAndGet(docSetSize);             
                HashMap<Utf8, Object> userDescMap = (HashMap<Utf8, Object>) docsRecord.get(GlobalParas.user_desc);
                file_id = Long.parseLong(userDescMap.get(new Utf8("file_id")).toString().split("[|]")[0]);
                log.info(new StringBuilder().append("get data topic file_id size ").append(getOriTopic(msg)).append(" ").append(file_id).append(" ").append(docSetSize).toString());
                if (GlobalParas.isTestMode) {
                    if (GlobalParas.id2Createindex.get(file_id) != null) {
                        idxCtrl = (IndexControler) GlobalParas.id2Createindex.get(file_id);
                        idxCtrl.inbuffer.put(docsRecord);
                    } else {
                        try {
                            Thread.sleep(2000L);
                        } catch (Exception e) {
                        }
                        log.info("in test mode,there is no file to wrtite,just sleep");
                    }
                } else {
                    idxCtrl = (IndexControler) GlobalParas.id2Createindex.get(Long.valueOf(file_id));
                    if (idxCtrl != null) {
                        if (!idxCtrl.isDiskBad.get()) {
                            if (idxCtrl.preEnd.get()) {
                                String information = new StringBuilder().append(file_id).append("|").append(getOriTopic(msg)).append("|||").append(GlobalParas.hostName).toString();
                                if (GlobalParas.delayErrorFileId.contains(Long.valueOf(file_id))) {
                                    if (System.currentTimeMillis() - ((Long) GlobalParas.delayErrorFileId.get(Long.valueOf(file_id))).longValue() >= GlobalParas.badfileinfoCheckGab * 1000) {
                                        RocketMQProducer.sendMessage(GlobalParas.error_file_mq, information.getBytes(), -1);
                                        GlobalParas.delayErrorFileId.put(Long.valueOf(file_id), Long.valueOf(System.currentTimeMillis()));
                                    }
                                    log.info(new StringBuilder().append("put back some data for delay error for ").append(information).toString());
                                    RocketMQProducer.sendMessage(getOriTopic(msg), msg.getBody(), docSetSize);
                                    continue;
                                }
                                log.info(new StringBuilder().append("has set preEnd to true,can not write file  for ").append(information).toString());
                                GlobalParas.delayErrorFileId.put(Long.valueOf(file_id), Long.valueOf(System.currentTimeMillis()));
                                RocketMQProducer.sendMessage(GlobalParas.error_file_mq, information.getBytes(), -1);
                                RocketMQProducer.sendMessage(getOriTopic(msg), msg.getBody(), docSetSize);
                                MetastoreWrapper.setFileBad(file_id);
                                continue;
                            }
                            while (!idxCtrl.inbuffer.offer(docsRecord, GlobalParas.indexerBufferFullSleepMilis, TimeUnit.MILLISECONDS)) {
                                log.warn(new StringBuilder().append("put data to buffer fail,will retry,file_id ").append(file_id).toString());
                            }
                            if (GlobalParas.isSendMetric) {
                                DataIndexerMetricReportor.sendMetric(new StringBuilder().append(GlobalParas.ip).append("_").append(System.nanoTime()).toString(), idxCtrl.tbName, idxCtrl.dbName, docSetSize, "normal", "in", "", "DP");
                            }
                        } else {
                            log.info(new StringBuilder().append("put back some disk bad data for  ").append(file_id).append("  ").append(msg.getTopic()).toString());
                            String information = new StringBuilder().append(file_id).append("|").append(getOriTopic(msg)).append("|||").append(GlobalParas.hostName).toString();
                            if (GlobalParas.diskErrorFileId.containsKey(Long.valueOf(file_id))) {
                                if (System.currentTimeMillis() - ((Long) GlobalParas.diskErrorFileId.get(Long.valueOf(file_id))).longValue() >= GlobalParas.badfileinfoCheckGab * 1000) {
                                    RocketMQProducer.sendMessage(GlobalParas.error_file_mq, information.getBytes(), -1);
                                    GlobalParas.diskErrorFileId.put(Long.valueOf(file_id), Long.valueOf(System.currentTimeMillis()));
                                }
                                log.info(new StringBuilder().append("put back data for disk error ").append(file_id).toString());
                                RocketMQProducer.sendMessage(getOriTopic(msg), msg.getBody(), docSetSize);
                                continue;
                            }
                            RocketMQProducer.sendMessage(GlobalParas.error_file_mq, information.getBytes(), -1);
                            RocketMQProducer.sendMessage(getOriTopic(msg), msg.getBody(), docSetSize);
                            GlobalParas.diskErrorFileId.put(Long.valueOf(file_id), Long.valueOf(System.currentTimeMillis()));
                            log.info(new StringBuilder().append("will set file bad for disk errors for ").append(file_id).toString());
                            idxCtrl.isFileStatusBad.set(true);
                            MetastoreWrapper.setFileBad(file_id);
                            log.info(new StringBuilder().append("set file bad for disk errors done for ").append(file_id).toString());
                            continue;
                        }
                    } else {
                        if (GlobalParas.diskErrorFileId.containsKey(Long.valueOf(file_id))) {
                            if (System.currentTimeMillis() - ((Long) GlobalParas.diskErrorFileId.get(Long.valueOf(file_id))).longValue() >= GlobalParas.badfileinfoCheckGab * 1000) {
                                String information = new StringBuilder().append(file_id).append("|").append(getOriTopic(msg)).append("|||").append(GlobalParas.hostName).toString();
                                RocketMQProducer.sendMessage(GlobalParas.error_file_mq, information.getBytes(), -1);
                                GlobalParas.diskErrorFileId.put(Long.valueOf(file_id), Long.valueOf(System.currentTimeMillis()));
                                log.info(new StringBuilder().append("put back data that is in disk error filelist ").append(information).toString());
                            }
                            RocketMQProducer.sendMessage(getOriTopic(msg), msg.getBody(), docSetSize);
                            continue;
                        }
                        if (GlobalParas.removedFileId.containsKey(Long.valueOf(file_id))) {
                            if (System.currentTimeMillis() - ((Long) GlobalParas.removedFileId.get(Long.valueOf(file_id))).longValue() >= GlobalParas.badfileinfoCheckGab * 1000) {
                                String information = new StringBuilder().append(file_id).append("|").append(getOriTopic(msg)).append("|||").append(GlobalParas.hostName).toString();
                                RocketMQProducer.sendMessage(GlobalParas.error_file_mq, information.getBytes(), -1);
                                GlobalParas.removedFileId.put(Long.valueOf(file_id), Long.valueOf(System.currentTimeMillis()));
                                log.info(new StringBuilder().append("put back data that is in removed filelist ").append(information).toString());
                            }
                            RocketMQProducer.sendMessage(getOriTopic(msg), msg.getBody(), docSetSize);
                            continue;
                        }
                        if (GlobalParas.metaORdeviceErrorFileId.containsKey(Long.valueOf(file_id))) {
                            if (System.currentTimeMillis() - ((Long) GlobalParas.metaORdeviceErrorFileId.get(Long.valueOf(file_id))).longValue() >= GlobalParas.badfileinfoCheckGab * 1000) {
                                String information = new StringBuilder().append(file_id).append("|").append(getOriTopic(msg)).append("|||").append(GlobalParas.hostName).toString();
                                RocketMQProducer.sendMessage(GlobalParas.error_file_mq, information.getBytes(), -1);
                                GlobalParas.metaORdeviceErrorFileId.put(Long.valueOf(file_id), Long.valueOf(System.currentTimeMillis()));
                                log.info(new StringBuilder().append("put back data that is in meta or wrong device error filelist ").append(information).toString());
                            }
                            RocketMQProducer.sendMessage(getOriTopic(msg), msg.getBody(), docSetSize);
                            continue;
                        }
                        bg = System.currentTimeMillis();
                        synchronized (this) {
                            log.info(new StringBuilder().append("get lock ").append(this).append(" use ").append(System.currentTimeMillis() - bg).append(" ms ").toString());
                            bg = System.currentTimeMillis();
                            if (GlobalParas.diskErrorFileId.containsKey(Long.valueOf(file_id))) {
                                if (System.currentTimeMillis() - ((Long) GlobalParas.diskErrorFileId.get(Long.valueOf(file_id))).longValue() >= GlobalParas.badfileinfoCheckGab * 1000) {
                                    String information = new StringBuilder().append(file_id).append("|").append(getOriTopic(msg)).append("|||").append(GlobalParas.hostName).toString();
                                    RocketMQProducer.sendMessage(GlobalParas.error_file_mq, information.getBytes(), -1);
                                    GlobalParas.diskErrorFileId.put(Long.valueOf(file_id), Long.valueOf(System.currentTimeMillis()));
                                    log.info(new StringBuilder().append("put back data that is in disk error filelist ").append(information).toString());
                                }
                                RocketMQProducer.sendMessage(getOriTopic(msg), msg.getBody(), docSetSize);
                                continue;
                            }
                            if (GlobalParas.removedFileId.containsKey(Long.valueOf(file_id))) {
                                if (System.currentTimeMillis() - ((Long) GlobalParas.removedFileId.get(Long.valueOf(file_id))).longValue() >= GlobalParas.badfileinfoCheckGab * 1000) {
                                    String information = new StringBuilder().append(file_id).append("|").append(getOriTopic(msg)).append("|||").append(GlobalParas.hostName).toString();
                                    RocketMQProducer.sendMessage(GlobalParas.error_file_mq, information.getBytes(), -1);
                                    GlobalParas.removedFileId.put(Long.valueOf(file_id), Long.valueOf(System.currentTimeMillis()));
                                    log.info(new StringBuilder().append("put back data that is in removed filelist ").append(information).toString());
                                }
                                RocketMQProducer.sendMessage(getOriTopic(msg), msg.getBody(), docSetSize);
                                continue;
                            }
                            if (GlobalParas.metaORdeviceErrorFileId.containsKey(Long.valueOf(file_id))) {
                                if (System.currentTimeMillis() - ((Long) GlobalParas.metaORdeviceErrorFileId.get(Long.valueOf(file_id))).longValue() >= GlobalParas.badfileinfoCheckGab * 1000) {
                                    String information = new StringBuilder().append(file_id).append("|").append(getOriTopic(msg)).append("|||").append(GlobalParas.hostName).toString();
                                    RocketMQProducer.sendMessage(GlobalParas.error_file_mq, information.getBytes(), -1);
                                    GlobalParas.metaORdeviceErrorFileId.put(Long.valueOf(file_id), Long.valueOf(System.currentTimeMillis()));
                                    log.info(new StringBuilder().append("put back back data that is in device error filelist ").append(information).toString());
                                }
                                RocketMQProducer.sendMessage(getOriTopic(msg), msg.getBody(), docSetSize);
                                continue;
                            }
                            try {
                                if (!GlobalParas.id2Createindex.containsKey(Long.valueOf(file_id))) {
                                    log.warn(new StringBuilder().append("the file has not been init in persi, but it is ok in metastore, get it from metastore: ").append(file_id).toString());
                                    StringBuilder sb = new StringBuilder();
                                    SFile sf = MetastoreWrapper.getFile(file_id, sb);
                                    if (sf == null) {
                                        String information = new StringBuilder().append(file_id).append("|").append(getOriTopic(msg)).append("|||").append(GlobalParas.hostName).toString();
                                        RocketMQProducer.sendMessage(GlobalParas.error_file_mq, information.getBytes(), -1);
                                        RocketMQProducer.sendMessage(getOriTopic(msg), msg.getBody(), docSetSize);
                                        GlobalParas.metaORdeviceErrorFileId.put(Long.valueOf(file_id), Long.valueOf(System.currentTimeMillis()));
                                        log.info(new StringBuilder().append("will set file bad for meta or device error(get file) for ").append(file_id).toString());
                                        MetastoreWrapper.setFileBad(file_id);
                                        log.info(new StringBuilder().append("set file bad for meta or device error(get file) done for ").append(file_id).toString());
                                        continue;
                                    }
                                    if (sf.getStore_status() != 0) {
                                        if ((sf.getStore_status() == 3) || (sf.getStore_status() == 4)) {
                                            String values = new StringBuilder().append("|").append(((SplitValue) sf.getValues().get(0)).getValue()).append(",").append(((SplitValue) sf.getValues().get(1)).getValue()).append("|").append(((SplitValue) sf.getValues().get(2)).getValue()).toString();
                                            String information = new StringBuilder().append(file_id).append("|").append(getOriTopic(msg)).append("|||").append(GlobalParas.hostName).toString();
                                            RocketMQProducer.sendMessage(GlobalParas.error_file_mq, information.getBytes(), -1);
                                            RocketMQProducer.sendMessage(getOriTopic(msg), msg.getBody(), docSetSize);
                                            log.warn(new StringBuilder().append("the data is not in increate status,it is: ").append(information).append(" ").append(sf).append(",the values are ").append(values).toString());
                                            GlobalParas.removedFileId.put(Long.valueOf(file_id), Long.valueOf(System.currentTimeMillis()));
                                            continue;
                                        }
                                        sb = new StringBuilder();
                                        if (MetastoreWrapper.reopenFile(file_id)) {
                                            sf = MetastoreWrapper.getFile(file_id, sb);
                                            log.info(new StringBuilder().append("now reopen the file ").append(file_id).append(" done,the file is ").append(sf).toString());
                                        } else {
                                            String information = new StringBuilder().append(file_id).append("|").append(getOriTopic(msg)).append("|||").append(GlobalParas.hostName).toString();
                                            RocketMQProducer.sendMessage(GlobalParas.error_file_mq, information.getBytes(), -1);
                                            RocketMQProducer.sendMessage(getOriTopic(msg), msg.getBody(), docSetSize);
                                            GlobalParas.metaORdeviceErrorFileId.put(Long.valueOf(file_id), Long.valueOf(System.currentTimeMillis()));
                                            log.info(new StringBuilder().append("will set file bad for meta or device error(reopen file) for ").append(information).append(",before reopen: ").append(sf).toString());
                                            MetastoreWrapper.setFileBad(file_id);
                                            log.info(new StringBuilder().append("set file bad for meta or device error(reopen file) done for ").append(information).append(",before reopen: ").append(sf).toString());
                                            continue;
                                        }
                                    } else {
                                        log.info(new StringBuilder().append("get one exist file ok,the file is ").append(sf.toString()).toString());
                                    }
                                    List luceneFileList = new ArrayList();
                                    String db = sf.getDbName();
                                    String tb = sf.getTableName();
                                    luceneFileList.add(sf);
                                    DevMap dm = new DevMap();
                                    String fullPath = "";
                                    SFileLocation location = null;
                                    try {
                                        for (SFileLocation tmpLocation : sf.getLocations()) {
                                            if (GlobalParas.hostName.equalsIgnoreCase(tmpLocation.getNode_name())) {
                                                location = tmpLocation;
                                                break;
                                            }
                                        }
                                        fullPath = dm.getPath(location.getDevid(), location.getLocation());
                                    } catch (Exception e) {
                                        log.error(e, e);
                                        String information = new StringBuilder().append(file_id).append("|").append(getOriTopic(msg)).append("|||").append(GlobalParas.hostName).toString();
                                        RocketMQProducer.sendMessage(GlobalParas.error_file_mq, information.getBytes(), -1);
                                        RocketMQProducer.sendMessage(getOriTopic(msg), msg.getBody(), docSetSize);

                                        GlobalParas.metaORdeviceErrorFileId.put(Long.valueOf(file_id), Long.valueOf(System.currentTimeMillis()));
                                        log.info(new StringBuilder().append("will set file bad for device error(get file) for ").append(file_id).append(" ").append(sf.toString()).toString());
                                        MetastoreWrapper.setFileBad(file_id);
                                        log.info(new StringBuilder().append("set bad done for device error(get file) for ").append(file_id).append(" ").append(sf.toString()).toString());
                                        continue;
                                    }
                                    log.info(new StringBuilder().append("will get the lucene file from disk,the fullpath is ").append(fullPath).toString()+" for file id "+file_id);
                                    String stTime = "";
                                    String edTime = "";
                                    for (int j = 0; j < 2; j++) {
                                        long time = Long.parseLong(((SplitValue) sf.getValues().get(j)).getValue().trim());
                                        if (j == 0) {
                                            stTime = String.valueOf(time);
                                        } else {
                                            edTime = String.valueOf(time);
                                        }
                                    }
                                    log.info(new StringBuilder().append("now set the time flag for file ").append(file_id).append(" to ").append(stTime).append("_").append(edTime).toString());
                                    DataSourceConfig ds = new DataSourceConfig(db, tb, stTime);
                                    ConcurrentHashMap hashWriterMap = new ConcurrentHashMap();
                                    ConcurrentHashMap luceneSFMap = new ConcurrentHashMap();
                                    try {
                                        IndexWriter FSWriter = LuceneWriterUtil.getSimpleLuceneWriter(fullPath, false, tb);
                                        FSWriter.close();
                                        FSWriter = LuceneWriterUtil.getSimpleLuceneWriter(fullPath, false, tb);
                                        log.info(new StringBuilder().append("get one lucene physicical file ok ->").append(file_id).append(";").append(Thread.currentThread().getName()).append("-").append(Thread.currentThread().getId()).toString());
                                        hashWriterMap.put(Integer.valueOf(0), FSWriter);
                                        luceneSFMap.put(Integer.valueOf(0), luceneFileList);
                                        LuceneFileWriter writer = new LuceneFileWriter(hashWriterMap, luceneSFMap);

                                        ArrayBlockingQueue buf = new ArrayBlockingQueue(GlobalParas.readbufSize);
                                        IndexControler c = new IndexControler(ds, buf, writer, file_id);
                                        c.threadPool = WorkerPool.hlwrzpool;
                                        Thread ct = new Thread(c);
                                        ct.setName(new StringBuilder().append(ds.getDbName()).append("_").append(ds.getTbName()).append("_").append(file_id).toString());
                                        ct.start();
                                        GlobalParas.id2Createindex.put(Long.valueOf(file_id), c);
                                        log.info(new StringBuilder().append("apply one file from metastore use  ").append(System.currentTimeMillis() - bg).append(" ms for ").append(file_id).toString());
                                    } catch (Exception e) {
                                        log.error(e, e);
                                        String information = new StringBuilder().append(file_id).append("|").append(getOriTopic(msg)).append("|||").append(GlobalParas.hostName).toString();
                                        RocketMQProducer.sendMessage(GlobalParas.error_file_mq, information.getBytes(), -1);
                                        RocketMQProducer.sendMessage(getOriTopic(msg), msg.getBody(), docSetSize);
                                        GlobalParas.metaORdeviceErrorFileId.put(Long.valueOf(file_id), Long.valueOf(System.currentTimeMillis()));
                                        log.info(new StringBuilder().append("will set file bad for meta or device error(get file) for ").append(information).append(" ").append(e).toString());
                                        MetastoreWrapper.setFileBad(file_id);
                                        log.info(new StringBuilder().append("set file bad for meta or device error(get file) done for ").append(information).append(" ").append(e).toString());
                                        continue;
                                    }
                                    log.info(new StringBuilder().append("check or new file for file id ").append(file_id).append(" use ").append(System.currentTimeMillis() - bg).append(" ms").toString());
                                }
                                while (!((IndexControler) GlobalParas.id2Createindex.get(Long.valueOf(file_id))).inbuffer.offer(docsRecord)) {
                                    Thread.sleep(GlobalParas.indexerBufferFullSleepMilis);
                                    log.warn(new StringBuilder().append("put data to buffer fail,will retry,file_id ").append(file_id).toString());
                                }
                                if (GlobalParas.isSendMetric) {
                                    DataIndexerMetricReportor.sendMetric(new StringBuilder().append(GlobalParas.ip).append("_").append(System.nanoTime()).toString(), ((IndexControler) GlobalParas.id2Createindex.get(Long.valueOf(file_id))).tbName, ((IndexControler) GlobalParas.id2Createindex.get(Long.valueOf(file_id))).dbName, docSetSize, "normal", "in", "", "DP");
                                }
                            } catch (Exception e) {
                                log.error(e, e);
                            }
                        }
                        log.info(new StringBuilder().append("release lock ").append(this).append(" and file operation use ").append(System.currentTimeMillis() - bg).append(" ms ").toString());
                    }
                }
            } catch (Exception ex) {
                log.error(ex, ex);
            }
        }
    }
}
